package com.zn.socket;

import com.zn.util.Constant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;

/**
 * Created by yinjiawei on 2018/7/12.
 */
public class ReaderTask extends Thread{
    private static final Logger LOGGER = LoggerFactory.getLogger(ReaderTask.class);
    private SocketStatusListener socketStatusListener;

//    private BufferedReader bufferedReader;

    private Socket socket;

    private boolean listening;

    private final ForwardingSocketMsg forwardingSocketMsg=new ForwardingSocketMsg();

    public ReaderTask(Socket socket) throws IOException
    {
//        bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
        this.socket = socket;
    }

    /**
     * finish:(这里用一句话描述这个方法的作用). <br/>
     * TODO(这里描述这个方法适用条件 – 可选).<br/>
     * @throws IOException
     *
     */
    public void finish() throws IOException
    {
        listening = false;
        interrupt();
        if(socket!=null)
        {
            if(socket.isInputShutdown())
            {
                socket.shutdownInput();
            }
        }
    }

    /* (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    @Override
    public synchronized  void run()
    {
        while (listening)
        {
            try {
                this.process();
            } catch (Exception e) {
                listening = false;
                if(socketStatusListener!=null)
                {
                    int status = parseSocketStatus(e);
                    socketStatusListener.onSocketStatusChanged(socket, status, e);
                }
                e.printStackTrace();
                return;//终止线程继续运行,这里也可以使用continue
            }

        }
    }

    public void process() throws Exception {
        // Prepare date format
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd HH:mm:ss.SSS");
        dateFormat.setTimeZone(Constant.DEFAULT_TIMEZONE);
        // Prepare read buffer
        ByteBuffer bufReceive = ByteBuffer.allocate(1024 * 200);
        byte bufRead[] = new byte[1024 * 100];
        InputStream is = socket.getInputStream();
        try {
            int count = 0;
            while((count = is.read(bufRead)) > 0) {
                if(count > bufReceive.remaining()) {
                    throw new Exception("Proxy data receiver buffer overflow");
                }
                bufReceive.put(bufRead, 0, count);
                this.parseData(bufReceive);
            }
        }
        finally {
            try {is.close();}catch(Exception e) {e.printStackTrace();}
            try {socket.close();}catch(Exception e) {e.printStackTrace();}
        }
    }

    /**
     * 解析交易数据
     * @param buf 数据缓冲
     */
    protected void parseData(ByteBuffer buf) {
        buf.flip();
        byte data[] = new byte[buf.remaining()];
        buf.get(data);
        int startIndex = 0;
        for(int i = 0; i < data.length; i++) {
            if(data[i] == '\n') {
                if(i > startIndex) {
                    String item = null;
                    try {
                        item = new String(data, startIndex, i - startIndex, "GB2312");
                        /**读取c++的报单消息的响应**/
                        if(item.startsWith("report")){
                            forwardingSocketMsg.receiverReportOrderMsg(item);
                        }
                        /**读取c++的成交消息的响应**/
                        if(item.startsWith("deal")){
                            forwardingSocketMsg.receiverDealOrderMsg(item);
                        }

                        /**读取给持仓单以及涨跌幅数据指令**/
                        if(item.startsWith("tradeStartUp")){
                            forwardingSocketMsg.receiverSendDataFlag(item);// 入口在这 c++会发送一个指令给我 然后我把数据通过tcp
                        }
                        /**读取风控平仓的消息响应**/
                        if(item.startsWith("riskCloseOrder")){
                            //forwardingSocketMsg.receiverRiskOrderMsg(item);
                        }
                        /**读取C++对报单引用校验的结果**/
                        if(item.startsWith("checkOrderRefResult")){
                            //forwardingSocketMsg.checkOrderRefResult(item);
                        }
                        /**将母账户信息给c++指令**/
                        if(item.startsWith("giveMeAccount")){
                            //forwardingSocketMsg.giveMeAccount(item);
                        }
                    }catch(Exception e) {
                        e.printStackTrace();
                        LOGGER.error(this.getClass()+"Proxy data receiver parse error: " , e);
                        LOGGER.error(this.getClass()+""+ item);
                    }
                }
                startIndex = i + 1;
            }
        }
        buf.clear();
        if(startIndex < data.length) {
            buf.put(data, startIndex, data.length - startIndex);
        }
    }

    private int parseSocketStatus(Exception e)
    {
        if(SocketException.class.isInstance(e))
        {
            String msg = e.getLocalizedMessage().trim();
            if("Connection reset".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_RESET;
            }
            else if("Socket is closed".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_CLOSE;
            }
            else if("Broken pipe".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_PIP_BROKEN;
            }

        }
        return SocketStatusListener.STATUS_UNKOWN;
    }

    /**
     * listen:(这里用一句话描述这个方法的作用). <br/>
     * TODO(这里描述这个方法适用条件 – 可选).<br/>
     *
     */
    public void startListener(SocketStatusListener ssl) {
        listening = true;
        this.socketStatusListener = ssl;
        start();
    }
}
